package com.ruoyi.device.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttDemo {

    /**
     * demo
     */
    public static void main(String[] args) {
        String subTopic = "testtopic/#";
        String pubTopic = "testtopic/1"; // 客户端发布消息的Topic名称
        String content = "Hello World"; // 测试的消息体
        int qos = 2;
        String broker = "tcp://192.168.125.10:1883"; // 服务端地址
        String clientId = "emqx_test"; // 客户端ID（唯一即可）
        MemoryPersistence persistence = new MemoryPersistence(); // 创建内存存储（持久化）

        try {
            // 创建mqtt客户端
            MqttClient client = new MqttClient(broker, clientId, persistence);
            // 连接emqx
            MqttConnectOptions connectOptions = new MqttConnectOptions();
            connectOptions.setUserName("emqx_test"); // 用户名
            connectOptions.setPassword("emqx_test_password".toCharArray()); // 密码
            // 保留会话
            connectOptions.setCleanSession(true);

            // 客户端连接服务器
            System.out.println("Connecting to broker: " + broker);
            client.connect(connectOptions);

            System.out.println("Connected");
            System.out.println("Publishing message: " + content);

            // 客户端演示 - 消息接受者
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    // 连接丢失后，一般在这里面进行重连
                    System.out.println("连接断开，可以做重连");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // subscribe后得到的消息会执行到这里面（接收到消息）
                    System.out.println("接收消息主题:" + topic);
                    System.out.println("接收消息Qos:" + message.getQos());
                    System.out.println("接收消息内容:" + new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // 推送完成
                    System.out.println("deliveryComplete---------" + token.isComplete());
                }
            });

            // 客户端订阅topic
            client.subscribe(subTopic);
            // 发送消息
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(2); // 设置消息质量为2(只有一次发送，确保消息只送达一次)，默认为0
            client.publish(pubTopic, message); // 发布消息
            System.out.println("Message published");
            // 关闭客户端连接
            client.disconnect();
            client.close();
            System.exit(0);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
